﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using AzureMRCore.DataTypes;
using Microsoft.WindowsAzure.StorageClient;

namespace AzureMRCore.OutputFormat
{
    public class DoubleMatrixOutputFormat : IOutputFormat<IntKey, MatrixValue<double>>
    {
        public Dictionary<string, IntKey> UploadValues(List<KeyValuePair<IntKey, MatrixValue<double>>> pairs,
                                                       object firstID, string iteration, CloudBlobContainer container)
        {
            Dictionary<string, IntKey> references = new Dictionary<string, IntKey>();
            MatrixValue<double> value = pairs[0].Value;

            string fileName = "Out_" + firstID + "_" + iteration + ".bin";
            CloudBlockBlob resultBlob = UploadMatrixValue(fileName, value, container);
            references.Add(resultBlob.Uri.AbsoluteUri, pairs[0].Key);
            return references;
        }

        public Dictionary<Stream, IntKey> GetStream(List<KeyValuePair<IntKey, MatrixValue<double>>> pairs)
        {
            Dictionary<Stream, IntKey> references = new Dictionary<Stream, IntKey>();
            MatrixValue<double> matrixValue = pairs[0].Value;
            MemoryStream stream = new MemoryStream();
            BinaryWriter writer = new BinaryWriter(stream);

            writer.Write(matrixValue.BlockHeight);
            writer.Write(matrixValue.Width);
            writer.Write(matrixValue.RowBlock);
            writer.Write(matrixValue.RowOffset);
            writer.Write(false);

            for (int j = 0; j < matrixValue.BlockHeight; j++)
            {
                for (int i = 0; i < matrixValue.Width; i++)
                {
                    writer.Write(matrixValue.Matrix[j][i]);
                }
            }

            writer.Flush();
            stream.Seek(0, SeekOrigin.Begin);
            references.Add(stream, pairs[0].Key);
            return references;
        }

        public long GetSize(List<KeyValuePair<IntKey, MatrixValue<double>>> pairs)
        {
            //disabled
            return long.MaxValue;
        }

        public byte[] ToBytes(List<KeyValuePair<IntKey, MatrixValue<double>>> pairs)
        {
            MatrixValue<double> matrixValue = pairs[0].Value;
            return matrixValue.GetBytes();
        }

        public void FromBytes(byte[] values, ConcurrentDictionary<IntKey, List<MatrixValue<double>>> decodedData)
        {
            throw new NotImplementedException();
        }

        private static CloudBlockBlob UploadMatrixValue(string blobName, MatrixValue<double> matrixValue,
                                                        CloudBlobContainer container)
        {
            MemoryStream stream = new MemoryStream();
            BinaryWriter writer = new BinaryWriter(stream);

            writer.Write(matrixValue.BlockHeight);
            writer.Write(matrixValue.Width);
            writer.Write(matrixValue.RowBlock);
            writer.Write(matrixValue.RowOffset);
            writer.Write(false);

            for (int j = 0; j < matrixValue.BlockHeight; j++)
            {
                for (int i = 0; i < matrixValue.Width; i++)
                {
                    writer.Write(matrixValue.Matrix[j][i]);
                }
            }

            writer.Flush();
            stream.Seek(0, SeekOrigin.Begin);
            CloudBlockBlob resultBlob = container.GetBlockBlobReference(blobName);
            resultBlob.UploadFromStream(stream);
            return resultBlob;
        }

        public void DecodeData(string fileName, string key,
                               ConcurrentDictionary<IntKey, List<MatrixValue<double>>> decodedData)
        {
            BinaryReader reader = new BinaryReader(File.Open(fileName, FileMode.Open));
            int height = reader.ReadInt32();
            int width = reader.ReadInt32();
            int row = reader.ReadInt32();
            int rowOffset = reader.ReadInt32();
            bool isKeyed = reader.ReadBoolean();
            double[][] matrix = new double[height][];

            for (int j = 0; j < height; j++)
            {
                matrix[j] = new double[width];
                for (var i = 0; i < width; i++)
                {
                    matrix[j][i] = reader.ReadDouble();
                }
            }
            var matrixValue = MatrixValue<double>.GetInstance(matrix, height, width, row, rowOffset);
            var intKey = IntKey.GetInstance(Int32.Parse(key));

            List<MatrixValue<double>> values = decodedData.GetOrAdd(intKey, new List<MatrixValue<double>>());
            lock (values)
            {
                values.Add(matrixValue);
            }
        }
    }
}